package com.semidata.trp.decision.frequency;

import java.util.HashMap;
import java.util.Map;

import com.semidata.rtd.core.datasource.redis.RedisDataSource;
import com.semidata.rtd.core.datasource.redis.RedisDataSourceService;
import com.semidata.rtd.core.log.Logger;
import com.semidata.rtd.core.service.ServiceManager;
import com.semidata.rtd.engine.job.AsyncTask;
import com.semidata.rtd.engine.job.Job;
import com.semidata.trp.decision.config.Config;
import com.semidata.trp.decision.job.DecisionJob;
import com.semidata.trp.frequency.Frequency;
import com.semidata.trp.frequency.FrequencyHandler;

public class TaskFrequency extends AsyncTask {

	private Map<String, Integer> results;

	public TaskFrequency(Job job, long timeout) {
		super(job, timeout);
	}

	@Override
	public void run() {
		final DecisionJob job = (DecisionJob) getJob();
		Logger.log("DEBUG", "start taskfrequency: check: "
		        + job.getJobContext().getRequest().getUserId());

		try {
			RedisDataSourceService service = ServiceManager
			        .getService(RedisDataSourceService.class);
			RedisDataSource datasource = service.getDataSource();
			Frequency fre = new Frequency(datasource);
			fre.setReadFrequencyHandler(new FrequencyHandler() {

				@SuppressWarnings("unchecked")
				@Override
				public void handler(Object o) {
					// TODO Auto-generated method stub
					if (o == null) {
						results = new HashMap<String, Integer>();
					} else {
						results = (Map<String, Integer>) o;
					}
					for (Map.Entry<String, Integer> m : results.entrySet())
						Logger.log("DEBUG", "key:" + m.getKey() + " value : "
						        + m.getValue());
					job.getJobContext().setFrequency(results);
					done();

				}
			});
			fre.readFrequency(job.getJobContext().getRequest().getUserId(), Config.getConfig().getTaskTimeout());
			Logger.log("DEBUG", "read frequencing");
		} catch (Exception e) {
			Logger.log("DEBUG", "error occurs when get frequency");
			Logger.log("ERROR", e);
			throw new RuntimeException(e);
		}

	}

}
